package ItemBased;

import java.io.IOException;
import java.util.Iterator;
import java.util.Vector;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class GenerateUserVector {

	
	public static class GenerateUserVectorMap
    extends Mapper<LongWritable, LongPairArrayWritable, LongWritable, LongPairWritable>{
	      
	    public void map(LongWritable key, LongPairArrayWritable value,
	    		OutputCollector<LongWritable, LongPairWritable> output, Reporter reporter
	                    ) throws IOException, InterruptedException {
	    	//normalize here
	    	
	    	for(Writable pair : value.get()){
	    		LongPairWritable p = (LongPairWritable) pair;
	    		output.collect(p.getFirst(), new LongPairWritable(key, p.getSecond()));
	    	}
	    	
	    }
		
	}	
	
	
	public static class GenerateUserVectorReduce
    extends Reducer<Text,IntWritable,Text,IntWritable> {
		
		public void reduce(LongWritable key, Iterator<LongPairWritable> values,
	    		OutputCollector<LongWritable, LongPairArrayWritable> output, Reporter reporter
	                    ) throws IOException, InterruptedException {
			Vector<LongPairWritable> list = new Vector<LongPairWritable> (); 
			while(values.hasNext()){
				list.add(values.next().get());
			}
			
			LongPairWritable[] pair = new LongPairWritable[list.size()];
			LongPairArrayWritable arrayPair = new LongPairArrayWritable();
			
			for(int i=0; i<list.size();i++){
				pair[i] = list.get(i);
			}
			arrayPair.set(pair);
			output.collect(key, arrayPair);
		}
	}	
}
